# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: jaewon

import copy
import json
import logging
import math
import os
from typing import List, Optional, Union

import requests
from recordclass import recordclass
from tornado import gen
from tornado.ioloop import PeriodicCallback

from coin.base.config import Config
from coin.base.param_util import to_list
from coin.base.timestamp import get_timestamp
from coin.exchange.base.fill_stats import FillStatsCalculator2
from coin.exchange.base.order_fill_util import FillManager
from coin.exchange.base.order_gateway import (BaseOrderGatewayInfoMutator,
                                              Order,
                                              OrderGatewayBase,
                                              OrderGatewayStatus,
                                              OrderType)
from coin.exchange.base.order_gateway_logger import gen_og_log_request
from coin.exchange.base.order_gateway_ready import PositionOrderGatewayReady
from coin.exchange.base.order_id_util import OrderIdGenerator
from coin.exchange.okex_futures.kr_rest.currency import OkexCurrency
from coin.exchange.okex_futures.kr_rest.futures_constants import (contract_type_list)
from coin.exchange.okex_futures.kr_rest.futures_product import (OkexFuturesProduct)
from coin.exchange.okex_futures.order_gateway_log import (OkexFuturesOrderGatewayLogProvider)
from coin.exchange.okex_futures.rest.private import OkexRestPrivate
from coin.exchange.okex_futures.types import OkexFuturesOrderSide
from coin.exchange.okex_futures.ws.trade import OkexFuturesTradeClient
from coin.exchange.util.rest_rate_limit import RestRateLimiter
from coin.proto.coin_order_gateway_pb2 import (OrderEvent, OrderGatewayConfig)
from coin.proto.coin_query_pb2 import (AccountBalance,
                                       AccountPosition,
                                       CurrencyBalance,
                                       ProductPosition)
from coin.strategy.hk_hard_arb.og_util import print_working_orders
# from coin.support.accounting.checker.account_checker import (
#     AccountChecker,
#     AccountCheckFailedError,
#     get_default_account_checker_config,
# )
# from coin.support.accounting.logic.account_status_util import Acc

OkexFuturesPosition = recordclass('OkexFuturesPosition',
                                  ['long', 'short', 'long_avg_price', 'short_avg_price'])

OkexFuturesOrderInternal = recordclass('OkexOrderInternal',
                                       [
                                           'lever_rate',
                                           'timestamp',
                                           'exchange_order_id',
                                           'is_external_order',
                                           'fill',
                                           'fully_filled',
                                           'cancel_confirmed',
                                           'cancel_sent_time',
                                           'status',
                                           'update_received_timestamp',
                                           'fee',
                                       ])


class AccountCheckFailedError(Exception):
  pass


def create_okex_futures_order_internal(*,
                                       lever_rate=None,
                                       timestamp=None,
                                       exchange_order_id=None,
                                       is_external_order=False,
                                       fill=None,
                                       fully_filled=False,
                                       cancel_confirmed=False,
                                       cancel_sent_time=None,
                                       status=None,
                                       update_received_timestamp=None,
                                       fee=0.0):
  return OkexFuturesOrderInternal(
      lever_rate=lever_rate,
      timestamp=timestamp,
      exchange_order_id=exchange_order_id,
      is_external_order=is_external_order,
      fill=fill,
      fully_filled=fully_filled,
      cancel_confirmed=cancel_confirmed,
      cancel_sent_time=cancel_sent_time,
      status=status,
      update_received_timestamp=update_received_timestamp,
      fee=fee,
  )


def _compare_order(order_1, order_2):
  return (order_1.product == order_2.product and math.isclose(order_1.price, order_2.price)
          and math.isclose(order_1.qty, order_2.qty) and order_1.order_side == order_2.order_side
          and order_1.order_type == order_2.order_type)


def create_okex_order_from_traderecord(data, timestamp):
  assert 'contract_name' in data, data
  native_symbol = data['contract_name']
  product = OkexFuturesProduct.FromStrNativeProduct(native_symbol)
  assert product.contract_type == data['contract_type']

  exchange_order_id = int(data['orderid'])
  price_orig = float(data['price'])
  price_avg_fill = float(data['price_avg'])
  qty_orig = float(data['amount'])
  qty_accum_fill = float(data['deal_amount'])
  lever_rate = int(data['lever_rate'])
  fee = float(data['fee'])

  future_order_side = OkexFuturesOrderSide(data['type'])
  order_type = OrderType.LIMIT

  status = int(data['status'])
  fill = FillManager(price_orig=price_orig,
                     qty_orig=qty_orig,
                     qty_accum_fill=qty_accum_fill,
                     price_avg_fill=price_avg_fill)
  internal = create_okex_futures_order_internal(lever_rate=lever_rate,
                                                exchange_order_id=exchange_order_id,
                                                fill=fill,
                                                status=status,
                                                update_received_timestamp=timestamp,
                                                fee=fee)

  order = Order(order_id=None,
                product=product,
                price=price_orig,
                qty=qty_orig,
                order_side=future_order_side,
                order_type=order_type,
                tag=None,
                accepted=True,
                cancel_sent=False,
                internal=internal)

  return status, order


def create_okex_order_from_orderinfo(data, timestamp):
  assert 'contract_name' in data, data
  native_symbol = data['contract_name']
  product = OkexFuturesProduct.FromStrNativeProduct(native_symbol)

  exchange_order_id = data['order_id']
  price_orig = float(data['price'])
  price_avg_fill = float(data['price_avg'])
  qty_orig = float(data['amount'])
  qty_accum_fill = float(data['deal_amount'])
  lever_rate = int(data['lever_rate'])
  fee = float(data['fee'])

  future_order_side = OkexFuturesOrderSide(int(data['type']))
  order_type = OrderType.LIMIT

  status = int(data['status'])
  fill = FillManager(price_orig=price_orig,
                     qty_orig=qty_orig,
                     qty_accum_fill=qty_accum_fill,
                     price_avg_fill=price_avg_fill)
  internal = create_okex_futures_order_internal(lever_rate=lever_rate,
                                                exchange_order_id=exchange_order_id,
                                                fill=fill,
                                                status=status,
                                                update_received_timestamp=timestamp,
                                                fee=fee)

  order = Order(order_id=None,
                product=product,
                price=price_orig,
                qty=qty_orig,
                order_side=future_order_side,
                order_type=order_type,
                tag=None,
                accepted=True,
                cancel_sent=False,
                internal=internal)

  return status, order


class OkexFuturesFillStatsCalculator(FillStatsCalculator2):
  ProductType = OkexFuturesProduct
  BuyOrderSide = [OkexFuturesOrderSide.BUY_OPEN, OkexFuturesOrderSide.BUY_CLOSE]
  SellOrderSide = [OkexFuturesOrderSide.SELL_OPEN, OkexFuturesOrderSide.SELL_CLOSE]

  def __init__(self, products):
    super().__init__(products)


class OkexFuturesGatewayInfo(BaseOrderGatewayInfoMutator):
  CurrencyType = OkexCurrency
  ProductType = OkexFuturesProduct
  BalanceType = float
  PositionType = float

  def __init__(self, currencies: List[OkexCurrency], products: List[OkexFuturesProduct]):
    super().__init__(currencies, products)
    self._margin_balance = {}

  def _update_margin_balance(self, currency: str, balance: float):
    self._margin_balance[currency] = balance

  # Margin balance includes unrealized pnl, but this is provided by exchange
  # and could be out of date or slow.
  # DO NOT USE THIS IN TRADING, ONLY ACCEPTABLE IN PNL DISPLAY.
  def get_margin_balance(self, currency: str):
    return self._margin_balance[currency]

  def initialized_to_zero(self):
    # Initialize to zero.
    for currency in self.currencies:
      currency_balance = CurrencyBalance(currency=currency.currency,
                                         currency_native=currency.native_currency,
                                         total=0.,
                                         available=0.,
                                         hold=0.)
      self.update_okex_futures_balance(currency, currency_balance)
      self._update_margin_balance(currency.currency, 0)

  def update_okex_futures_balance(self, currency, currency_balance: CurrencyBalance):
    self.set_balance(currency, currency_balance.total, currency_balance)

  def update_okex_futures_position(self, product, okex_futures_position: OkexFuturesPosition):
    net_position = okex_futures_position.long - okex_futures_position.short
    self.set_position(product, net_position, okex_futures_position)

  def balance_as_proto(self):
    return AccountBalance(exchange='Okex',
                          market_type='Futures',
                          each_balance=self.get_notNone_balances_internal())

  def position_as_proto(self):
    product_positions = []
    for symbol, position in self.get_positions().items():
      okex_futures_position = self._positions_internal[symbol]
      product = OkexFuturesProduct.FromStr(symbol)
      if okex_futures_position is not None:
        product_position = ProductPosition(symbol=product.symbol,
                                           native_symbol=product.native_symbol,
                                           long_position=okex_futures_position.long,
                                           short_position=okex_futures_position.short,
                                           net_position=position)
      else:
        product_position = ProductPosition(symbol=product.symbol,
                                           native_symbol=product.native_symbol,
                                           long_position=0,
                                           short_position=0,
                                           net_position=0)
      product_positions.append(product_position)
    return AccountPosition(exchange='Okex', market_type='Futures', each_position=product_positions)

  def get_okex_futures_position(self, product: OkexFuturesProduct) \
      -> Optional[OkexFuturesPosition]:
    if product.symbol not in self._positions_internal:
      return None
    return self._positions_internal[product.symbol]

  def _update_balance(self, currency: str, balance: dict, from_res=False):
    hold = float(balance['keep_deposit'])
    if from_res:
      margin_balance = float(balance['account_rights'])
      # account_rights includes realized+unrealized pnl.
      balance = (float(balance['account_rights']) - float(balance['profit_unreal']))
    else:
      balance = float(balance['balance']) + float(balance['profit_real'])

    # Note that |balance| does not include unrealized pnl.
    # TODO(jaewon): Add unrealized pnl to balance

    if balance == 0 and currency not in self.get_balances():
      return

    currency = OkexCurrency.FromStr(currency)
    currency_balance = CurrencyBalance(currency=currency.currency,
                                       currency_native=currency.native_currency,
                                       total=balance,
                                       available=(balance - hold),
                                       hold=hold)
    self.update_okex_futures_balance(currency, currency_balance)
    if from_res:
      self._update_margin_balance(currency.currency, margin_balance)

  def _update_balances_by_userinfo(self, data):
    self.initialized_to_zero()
    balance_infos = data['info']
    for currency, balance in balance_infos.items():
      self._update_balance(currency, balance, from_res=True)

  def _update_position_by_positioninfo(self, data):
    position_infos = data['positions']
    for position in position_infos:
      native_symbol = position['contract_name']
      product = OkexFuturesProduct.FromStrNativeProduct(native_symbol)
      position_long_or_short = ('long' if str(position['position']) == '1' else 'short')
      self._update_position(product, position_long_or_short, position)

  def _update_position(self,
                       product: OkexFuturesProduct,
                       position_long_or_short: str,
                       position: Union[int, float, dict],
                       avg_price: Optional[float] = None):
    if isinstance(position, dict):
      assert avg_price is None
      hold_amount = float(position['hold_amount'])
      avg_price = float(position['avgprice']) if hold_amount != 0 else 0.
      '''TODO(jaewon): What are these fields for?
      eveningup = position['eveningup']
      margin = position['margin']
      '''
    else:
      hold_amount = float(position)
      avg_price = avg_price if hold_amount != 0 else 0.

    pos = self.get_okex_futures_position(product)
    if pos is None:
      return

    if position_long_or_short == 'long':
      pos.long = hold_amount
      pos.long_avg_price = avg_price
    elif position_long_or_short == 'short':
      pos.short = hold_amount
      pos.short_avg_price = avg_price
    else:
      raise ValueError('Invalid position direction: %s' % position_long_or_short)

    self.update_okex_futures_position(product, pos)

  def _update_position2(self, product, position):
    # Initialize position.
    self.update_okex_futures_position(product, OkexFuturesPosition(0., 0., 0., 0.))

    if position is None or not position['holding']:
      self._update_position(product, 'long', 0., 0.)
      self._update_position(product, 'short', 0., 0.)
      return

    assert position['result']
    assert len(position['holding']) == 1
    pos = position['holding'][0]

    product_from_response = OkexFuturesProduct.FromCurrencyPairAndContractType(
        pos['symbol'], pos['contract_type'])
    assert product == product_from_response

    long_pos = float(pos['buy_amount'])
    long_avg_price = float(pos['buy_price_avg']) if long_pos != 0 else 0.
    short_pos = float(pos['sell_amount'])
    short_avg_price = float(pos['sell_price_avg']) if short_pos != 0 else 0.
    self._update_position(product, 'long', long_pos, long_avg_price)
    self._update_position(product, 'short', short_pos, short_avg_price)


class OkexFuturesOrderGateway(OrderGatewayBase):
  _exchange = 'Okex'
  _market_type = 'Futures'
  _api_version = None

  OGInfoType = OkexFuturesGatewayInfo

  def __init__(self,
               currencies: List[OkexCurrency],
               products: Union[List[OkexFuturesProduct], OkexFuturesProduct],
               config: Config,
               og_config=None,
               account_check_config=None,
               logger=None):
    super().__init__()

    self._logger = logger or logging.getLogger(__name__)

    self._og_ready_rest = PositionOrderGatewayReady()
    self._og_ready = PositionOrderGatewayReady()
    self._og_info_rest = OkexFuturesGatewayInfo(currencies, to_list(products))
    self._og_info = OkexFuturesGatewayInfo(currencies, to_list(products))
    self._og_logger = OkexFuturesOrderGatewayLogProvider(gen_og_log_request(config, self))

    self._working_orders = {}
    self._orders_by_exchange_order_id = {}

    self._ioloop = None
    self._ws_client = OkexFuturesTradeClient(config.api_key, config.api_secret, logger=self._logger)
    self._ws_client.ready_callback = self._on_ws_client_ready
    self._ws_client.close_callback = self._on_ws_client_close
    self._ws_client.message_callback = self._on_ws_client_message

    self._rest_client = OkexRestPrivate(config.api_key, config.api_secret)
    self._rest_rate_limiter = RestRateLimiter(delay=1)  # sec

    self._order_id_generator = OrderIdGenerator()
    self._order_counter_success = 0

    self._og_config = og_config or OrderGatewayConfig()

    self._account_snapshot_periodic_queryer = None
    self._querying_account = False

    self._orders_snapshot_periodic_queryer = None
    self._querying_working_orders = False

    # self._account_check_config = account_check_config or get_default_account_checker_config(
    #     self.og_info.currencies, self.og_info.products)
    # self._account_checker = AccountChecker(self._account_check_config)
    self._account_checker_failed = False
    self._periodic_account_checker = None

    self._fill_stats_calculator = OkexFuturesFillStatsCalculator(products)

    self._last_balance_update_timestamp = None
    self._last_order_info_update_timestamp = None
    self._balance_update_checker = None

    self._logger.info('og_config:\n%s', str(self._og_config))
    # self._logger.info('account_check_config:\n%s', str(self._account_check_config))

  def _periodic_balance_update_check(self):
    if self._last_order_info_update_timestamp is None:
      return

    time_diff = abs(self._last_order_info_update_timestamp - self._last_balance_update_timestamp)

    if time_diff > 30 * 1e9:
      self._on_ws_client_close()

  def _periodic_balance_update_check_install(self):
    self._balance_update_checker = PeriodicCallback(self._periodic_balance_update_check, 10000)
    self._balance_update_checker.start()

  def get_fill_stats(self, product):
    return self._fill_stats_calculator.get_fill_stats(product)

  def is_og_info_in_sync(self):
    return self._og_info == self._og_info_rest

  def _initialize_og_info(self):
    # balance
    # is updated by websocket, _on_userinfo_res
    assert self._og_ready.balance_ready, self._og_ready.balance_ready

    # position
    # must be updated by rest query
    for product in self.og_info.products:
      self._og_info.set_position(
          product,
          self._og_info_rest.get_position(product),
          self._og_info_rest.get_okex_futures_position(product),
      )
    self._og_ready.set_position_ready(True)

    # orders
    # is updated by websocket
    assert self._og_ready.orders_ready, self._og_ready.orders_ready

    self._og_ready.check_ready()
    self._og_ready_rest.check_ready()

  def _debug_print_working_orders(self):
    print_working_orders(self._logger, self)

  def get_okex_futures_position(self, product: OkexFuturesProduct) \
      -> Optional[OkexFuturesPosition]:
    return self._og_info.get_okex_futures_position(product)

  def _add_foreign_order(self, status, order):
    assert isinstance(order.internal.exchange_order_id, int)
    timestamp = get_timestamp()
    order.order_id = self._order_id_generator.gen_order_id(is_foreign=True)
    order.internal.timestamp = timestamp
    order.internal.update_received_timestamp = timestamp
    self._working_orders[order.order_id] = order
    self._orders_by_exchange_order_id[order.internal.exchange_order_id] = order

  def _generate_foreign_order(self, status, order_info):
    order_id = self._order_id_generator.gen_order_id(is_foreign=True)
    fill = FillManager(price_orig=order_info.price,
                       qty_orig=order_info.qty,
                       qty_accum_fill=0,
                       price_avg_fill=0)
    internal = copy.copy(order_info.internal)
    internal.fill = fill
    order = copy.copy(order_info)
    order.order_id = order_id
    order.internal = internal
    self._working_orders[order_id] = order
    self._orders_by_exchange_order_id[order.internal.exchange_order_id] = order
    return order

  def _update_order(self, status, order):
    # TODO(inkyu): change order => order_info / prev_order => order
    if status not in (-1, 0, 1, 2, 3, 4, 5):  # 3 also means cancel in progress.
      self._logger.error("Unknown status code: %s. Order: %s", status, order)
      return

    prev_order = None
    exchange_order_id = order.internal.exchange_order_id
    if exchange_order_id in self._orders_by_exchange_order_id:
      prev_order = self._orders_by_exchange_order_id[exchange_order_id]

    if prev_order is not None and (prev_order.internal.cancel_confirmed
                                   or prev_order.internal.fully_filled):
      if prev_order.internal.status not in (-1, 2):  # TODO(inkyu): This happens. Why?
        self._logger.error('Weird status: %s, %s', prev_order, order)

      if status == -1:
        if prev_order.internal.status == 2:
          self._logger.warning('Receive cancel confirmation after order is fully filled.')
        else:
          # Receive duplicated cancel confirmation is normal, ignore.
          pass
        return
      elif status == 0:
        self._logger.warning(
            'Order is already dead (status=%s), but we received pending order message. %s, fill = %s',
            prev_order.internal.status,
            prev_order,
            prev_order.internal.fill.value)
        return
      elif status == 1:
        if prev_order.internal.status == -1:
          self._logger.warning(
              'Order is already cancel confirmed, but we received partial fill message. %s, fill = %s',
              prev_order,
              prev_order.internal.fill.value)
          # Do not return, proceed to process this message.
        else:
          self._logger.warning(
              'Received fill message is out of order, fully filled message is received before partial fill message.'
          )
          return
      elif status == 2:
        if prev_order.internal.status == -1:
          self._logger.warning(
              'Received fully-filled message after cancel confirmation. %s, fill = %s',
              prev_order,
              prev_order.internal.fill.value)
          # Process to process this message
        else:
          self._logger.warning('Duplicated fully-filled message, %s, fill = %s',
                               prev_order,
                               prev_order.internal.fill.value)
          return
      elif status in (3, 4, 5):  # No need to handle
        return

    timestamp = get_timestamp()

    if prev_order is None:
      # Order submission response has not been received, or it is a foreign order.
      prev_order = self._generate_foreign_order(status, order)
      self._logger.info('Accpeted foreign order: %s', prev_order.order_id)

    assert _compare_order(prev_order, order), '%s != %s' % (prev_order, order)
    prev_order.internal.update_received_timestamp = order.internal.update_received_timestamp
    prev_order.internal.status = status
    # status: https://www.okex.com/ws_api.html
    #   -1: cancelled
    #    0: pending
    #    1: partially filled
    #    2: fully filled
    #    3, 4: cancel request in process
    #    5, cancel an already canceled order.
    # TODO(xguo) use enum?
    if status == -1:  # CANCELED
      # generate cancel confirmed
      self._og_logger.gen_order_event_and_log(status,
                                              prev_order,
                                              timestamp,
                                              OrderEvent.CANCEL_CONFIRMED)
      if prev_order.order_id in self._working_orders:
        del self._working_orders[prev_order.order_id]
      prev_order.internal.cancel_confirmed = True
    elif status == 0:  # PENDING
      # same as submit response
      pass  # handled already
    elif status in (1, 2):  # FILL
      last_fill = prev_order.internal.fill.update_by_accum_fill(
          order.internal.fill.value.price_avg_fill, order.internal.fill.value.qty_accum_fill)
      if last_fill is not None:
        assert last_fill.qty_last_fill > 0 and last_fill.price_last_fill > 0, last_fill
        self._fill_stats_calculator.add_fill_by_order(product=order.product,
                                                      filled_qty=abs(last_fill.qty_last_fill),
                                                      order_side=order.order_side,
                                                      order_qty=order.qty)
        self._og_logger.gen_order_event_and_log(status,
                                                prev_order,
                                                timestamp,
                                                OrderEvent.ORDER_FILLED,
                                                last_fill.qty_last_fill)
        prev_order.internal.fill.log_last_fill(prev_order, self._logger)
        self._publish_fill(prev_order.internal.fill.value.price_last_fill,
                           prev_order.internal.fill.value.qty_last_fill,
                           order)
      if status == 2:
        if not prev_order.internal.fill.fully_filled:
          self._logger.warning('Get fully filled status code, but the order is not fully filled.',
                               order)
        if prev_order.order_id in self._working_orders:
          del self._working_orders[prev_order.order_id]
        prev_order.internal.fully_filled = True
    elif status in (3, 4):  # CANCEL IN PROGRESS
      self._og_logger.gen_order_event_and_log(status,
                                              prev_order,
                                              timestamp,
                                              OrderEvent.CANCEL_ACCEPTED)
    elif status == 5:  # ORDER CANCELED ALREADY
      self._logger.info('Order %s canceled already (status: %s)', prev_order.order_id, status)

  @gen.coroutine
  def _query_position_impl(self):
    for product in self._og_info.products:
      for i in range(0, 5):
        try:
          response = yield self._rest_client.position(product.order_symbol, product.contract_type)
          if response.status_code != 200:
            self._logger.error('Invalid status: %s', response)
          else:
            message = json.loads(response.content)
            if not message['result']:  # Try again.
              continue
            self._og_info_rest._update_position2(product, message)
            break
        except requests.exceptions.RequestException:
          self._logger.exception('RequestException raised. Retrying in 5 seconds')
          yield gen.sleep(5)
        except json.JSONDecodeError:
          self._logger.exception('JsonDecodeError')
        except KeyError:
          self._logger.exception('KeyError')
        except Exception:
          self._logger.exception('Error')
          return
      else:
        self._logger.error('Tried %d times but failed. Assuming 0 position: %s',
                           i + 1,
                           str(product))
        self._og_info_rest._update_position2(product, None)

    self._og_ready_rest.set_position_ready(True)

  @gen.coroutine
  def _query_balance_impl(self):
    for i in range(0, 5):
      try:
        response = yield self._rest_client.userinfo()
        if response.status_code != 200:
          self._logger.error('Invalid status: %s', response)
        else:
          message = json.loads(response.content)
          if not message['result']:  # Try again.
            continue
          self._og_info_rest._update_balances_by_userinfo(message)
          break
      except requests.exceptions.RequestException:
        self._logger.exception('RequestException raised. Retrying in 5 seconds')
        yield gen.sleep(5)
      except json.JSONDecodeError:
        self._logger.exception('JsonDecodeError')
      except KeyError:
        self._logger.exception('KeyError')
      except Exception:
        self._logger.exception('Error')
        return
    else:
      self._logger.error('Tried %d times but failed.', i + 1)

    self._og_ready_rest.set_balance_ready(True)

  @gen.coroutine
  def _query_working_orders_initial_impl(self):
    for product in self._og_info.products:
      for i in range(0, 5):
        try:
          response = yield self._rest_client.order_history(product.order_symbol,
                                                           product.contract_type,
                                                           working_order=True)
          if response.status_code != 200:
            self._logger.error('Invalid status: %s', response)
          else:
            message = json.loads(response.content)
            if not message['result']:  # Try again.
              continue
            self._on_rest_client_order_history(message)
            break
        except requests.exceptions.RequestException:
          self._logger.exception('RequestException raised. Retrying in 5 seconds')
          yield gen.sleep(5)
        except json.JSONDecodeError:
          self._logger.exception('JsonDecodeError')
        except AssertionError:
          self._logger.exception('AssertionError')
        except Exception:
          self._logger.exception('Error')
          return
      else:
        self._logger.error('Tried %d times but failed. No quering working orders' % (i + 1))

    self._og_ready.set_orders_ready(True)
    self._og_ready_rest.set_orders_ready(True)

  @gen.coroutine
  def _query_working_orders_impl(self):
    for product in self._og_info.products:
      try:
        response = yield self._rest_client.order_history(product.order_symbol,
                                                         product.contract_type,
                                                         working_order=True)
        if response.status_code != 200:
          self._logger.error('Invalid status: %s', response)
        else:
          order_history = json.loads(response.content)
          self._on_rest_client_order_history(order_history)
      except json.JSONDecodeError:
        self._logger.exception('JsonDecodeError')
      except requests.exceptions.RequestException:
        self._logger.exception('RequestException')
      except Exception as e:
        self._logger.exception('Error: %s' % e.__class__)

    self._og_ready.set_orders_ready(True)
    self._og_ready_rest.set_orders_ready(True)

  def submittable(self):
    return self.is_ready() and self._rest_rate_limiter.is_available()

  @gen.coroutine
  def _initialize_order_gateway(self):
    yield self._query_balance_impl()
    yield self._query_position_impl()
    yield self._query_working_orders_initial_impl()

    self._logger.info('Position:\n%s',
                      json.dumps(self._og_info.get_positions(), sort_keys=True, indent=2))

    if self._working_orders:
      working_orders_str = '\n  '.join([str(order) for order in self._working_orders.values()])
      self._logger.info('Working orders:\n  %s' % working_orders_str)
    else:
      self._logger.info('No working orders')

    self._periodic_query_install()
    self._initialize_og_info()

    for _ in range(180):
      if self.is_og_info_in_sync():
        break
      else:
        self._logger.info('Wait until ws and rest order_infos are in sync.')
        yield gen.sleep(1)

    can_continue = True

    if not (self._og_ready.is_ready() and self._og_ready_rest.is_ready()):
      self._logger.error('OkexFutures OG ws/rest og_infos not ready at initialization stage!')
      can_continue = False

    if not self.is_og_info_in_sync():
      self._logger.error('OkexFutures OG ws/rest og_infos not in sync at initialization stage!')
      can_continue = False

    if not can_continue:
      self._logger.error("rest balance:\n%s", self._og_info_rest.balance_as_proto())
      self._logger.error("ws balance:\n%s", self._og_info.balance_as_proto())
      self._logger.error("rest position:\n%s", self._og_info_rest.position_as_proto())
      self._logger.error("ws position:\n%s", self._og_info.position_as_proto())
      os._exit(1)

    self._og_ready.set_status(OrderGatewayStatus.READY)
    self._logger.info('OkexFutures OG ready')

    self._periodic_check_install()
    self._periodic_balance_update_check_install()

  def _periodic_query_install(self):
    self._account_snapshot_periodic_queryer = PeriodicCallback(
        self._periodic_query_account, self._og_config.account_query_period * 1000)
    self._account_snapshot_periodic_queryer.start()

    self._orders_snapshot_periodic_queryer = PeriodicCallback(
        self._periodic_query_working_orders, self._og_config.orders_snapshot_query_period * 1000)
    self._orders_snapshot_periodic_queryer.start()

  def _periodic_check_install(self):
    self._periodic_account_checker = PeriodicCallback(self._periodic_check_og_info,
                                                      self._og_config.account_query_period
                                                      * 1000)  # same frequency as account query
    self._periodic_account_checker.start()

  @gen.coroutine
  def _periodic_check_og_info(self):
    try:
      account_balance_ws = self._og_info.balance_as_proto()
      account_position_ws = self._og_info.position_as_proto()
      account_balance_rest = self._og_info_rest.balance_as_proto()
      account_position_rest = self._og_info_rest.position_as_proto()

      self._account_checker.check_balances_by_range(account_balance_ws)
      self._account_checker.check_balances_by_range(account_balance_rest)
      self._account_checker.check_positions_by_range(account_position_ws)
      self._account_checker.check_positions_by_range(account_position_rest)
    except AccountCheckFailedError:
      self._logger.exception('OkexFutures OG account range check failed!')
      self._logger.error('balance or position in og_info is not reasonable!')
      self._logger.error("rest balance:\n%s", self._og_info_rest.balance_as_proto())
      self._logger.error("ws balance:\n%s", self._og_info.balance_as_proto())
      self._logger.error("rest position:\n%s", self._og_info_rest.position_as_proto())
      self._logger.error("ws position:\n%s", self._og_info.position_as_proto())
      self._account_checker_failed = True

  @gen.coroutine
  def _periodic_query_account(self):
    if self._querying_account:
      return
    self._querying_account = True
    yield self._query_balance_impl()
    yield self._query_position_impl()
    self._querying_account = False

  @gen.coroutine
  def _periodic_query_working_orders(self):
    if self._querying_working_orders:
      return
    self._querying_working_orders = True
    yield self._query_working_orders_impl()
    self._querying_working_orders = False

  def _on_rest_client_order_history(self, order_history):
    assert order_history['result'], order_history
    timestamp = get_timestamp()
    for orderinfo in order_history.get('orders', []):
      status, order = create_okex_order_from_orderinfo(orderinfo, timestamp)
      if order.product not in self._og_info.products:
        continue
      if order.internal.exchange_order_id not in self._orders_by_exchange_order_id:
        self._logger.info('Adding a foreign order: %s', order)
        self._add_foreign_order(status, order)

  def _on_cancel_order_res(self, msg):
    try:
      if msg['data']['result']:
        exchange_order_id = int(msg['data']['order_id'])
        if exchange_order_id in self._orders_by_exchange_order_id:
          order = self._orders_by_exchange_order_id[exchange_order_id]
          self._update_order(-1, order)
        else:
          self._logger.error('Receive cancel confirmation of unknown order %s', exchange_order_id)
    except (KeyError, TypeError):
      self._logger.exception('Cancel order error: %s', msg)

  def _on_ws_client_ready(self):
    self._ws_client.user_info()

  def _on_ws_client_close(self):
    self._ws_client.stop(trigger_close_callback=False)
    self._og_ready.set_reinit()
    self._og_ready_rest.set_reinit()
    self._og_info.reset()
    self._og_info_rest.reset()
    self._og_ready.set_status(OrderGatewayStatus.DISCONNECTED)
    self._og_ready_rest.set_status(OrderGatewayStatus.DISCONNECTED)
    self._balance_update_checker.stop()
    self._periodic_account_checker.stop()
    self._account_snapshot_periodic_queryer.stop()
    self._orders_snapshot_periodic_queryer.stop()

    self._logger.error('Trade client disconnected, will reconnect soon.')
    self._ws_client = self._ws_client.clone()  # create a new one.
    self._ws_client.start(self._ioloop)

  def _on_ws_client_message(self, msg):
    channel = msg.get('channel', '')

    if channel == 'ok_sub_futureusd_trades':
      self._last_order_info_update_timestamp = get_timestamp()
      self._on_trades_sub(msg)

    elif channel == 'ok_sub_futureusd_userinfo':
      self._last_balance_update_timestamp = get_timestamp()
      self._on_balance_sub(msg)

    elif channel == 'ok_sub_futureusd_positions':
      self._on_position_sub(msg)

    elif channel == 'ok_futureusd_userinfo':
      self._last_balance_update_timestamp = get_timestamp()
      self._on_userinfo_res(msg)

    elif channel == 'ok_futureusd_orderinfo':
      self._on_orderinfo_res(msg)

    elif channel == 'ok_futureusd_cancel_order':
      self._on_cancel_order_res(msg)

    elif channel == 'ok_futureusd_trade':
      # uses rest to submit order. do not use.
      pass

    elif channel.startswith('ok_spot_'):
      pass

    elif channel.startswith('ok_sub_spot_'):
      pass

    else:
      self._logger.info("unhandled message: %s, %s", channel, msg)

  def _on_position_sub(self, msg):
    self._og_info._update_position_by_positioninfo(msg['data'])
    self._og_ready.set_position_ready(True)
    self._og_logger.write_position(self._og_info.position_as_proto())

  def _on_userinfo_res(self, msg):
    data = msg['data']
    if not data['result']:
      self._logger.error('Error getting userinfo. Trying again. Message: %s', msg)
      self._ws_client.user_info()
      return

    self._og_info._update_balances_by_userinfo(data)
    self._og_ready.set_balance_ready(True)

    self._og_logger.write_balance(self._og_info.balance_as_proto())
    self._logger.info('Balance:\n%s',
                      json.dumps(self._og_info.get_balances(), sort_keys=True, indent=2))

    self._ioloop.add_callback(self._initialize_order_gateway)

  def _on_trades_sub(self, msg):
    traderecord = msg['data']
    timestamp = get_timestamp()
    status, order = create_okex_order_from_traderecord(traderecord, timestamp)
    if order.product not in self._og_info.products:
      return
    self._update_order(status, order)

  def _on_orderinfo_res(self, msg):
    data = msg['data']
    timestamp = get_timestamp()
    for orderinfo in data.get('orders', []):
      status, order = create_okex_order_from_orderinfo(orderinfo, timestamp)
      if order.product not in self._og_info.products:
        continue

      self._update_order(status, order)

  def _on_balance_sub(self, msg):
    balance_info = msg['data']
    currency, _ = balance_info['symbol'].split('_')
    self._og_info._update_balance(currency, balance_info)
    self._og_ready.set_balance_ready(True)

    self._og_logger.write_balance(self._og_info.balance_as_proto())

  def _remove_unacked_order(self, order_id, timestamp):
    order = self._working_orders[order_id]
    self._og_logger.gen_flush_unacked_order_event_and_log(order.product.symbol,
                                                          order.order_id,
                                                          timestamp)
    self._working_orders.pop(order_id, None)
    self._logger.warning(('Order %s is un-acked more than %s seconds. '
                          'Removing from on-the-fly order list.'),
                         str(order),
                         self._og_config.unacked_order_timeout)
    self._logger.error('Should never reach here!')

  def _cancel_order_with_no_updates(self, order_id):
    order = self._working_orders[order_id]
    self.cancel(order_id, ignore_error=True)
    self._logger.warning('Order %s is has no updates for more than %s seconds. cancel it.',
                         order,
                         self._og_config.unacked_order_timeout)

  def clean_old_unacked_orders(self):
    self._logger.debug('Cleaning old un-acked orders.')
    now_ts = get_timestamp()

    unacked_timeout = self._og_config.unacked_order_timeout * 1e9

    for order in self.get_working_order():
      submit_ts = order.internal.timestamp or 0.0
      if order.accepted:
        update_received_ts = order.internal.update_received_timestamp or 0.0
        if now_ts - update_received_ts > 2 * unacked_timeout and not order.cancel_sent:
          self._cancel_order_with_no_updates(order.order_id)
      elif now_ts - submit_ts > unacked_timeout:
        # Remove unacked orders. (Should not happen.)
        self._remove_unacked_order(order.order_id, now_ts)

    cancelled_order_id_list = [
        order.order_id for order in self._working_orders.values() if order.cancel_sent
    ]

    for order_id in cancelled_order_id_list:
      order = self._working_orders[order_id]
      cancel_ts = order.internal.cancel_sent_time
      if not cancel_ts:
        self._logger.error('cancel_sent_time is not set! This should not happen. %s',
                           self._working_orders[order_id])
        continue

      timeout = self._og_config.unacked_cancel_timeout * (10**9)
      if timeout < now_ts - cancel_ts <= 2 * timeout:
        self._logger.info('Cancel elapsed on %s is %s', order_id, now_ts - cancel_ts)
        exchange_order_id = order.internal.exchange_order_id
        self._ws_client.orderinfo(order.product.order_symbol,
                                  order.product.contract_type,
                                  exchange_order_id)
      elif now_ts - cancel_ts > 2 * timeout:
        self._logger.info('Cancel elapsed on %s is %s. retry cancel', order_id, now_ts - cancel_ts)
        exchange_order_id = order.internal.exchange_order_id
        self._ws_client.orderinfo(order.product.order_symbol,
                                  order.product.contract_type,
                                  exchange_order_id)
        self.cancel(order_id, ignore_error=True)

  def start(self, queue, ioloop, flow_sub=None):
    self._ioloop = ioloop
    self._og_ready.set_init()
    self._og_ready_rest.set_init()
    self._ws_client.start(ioloop)

    self._unacked_order_periodic_cleaner = PeriodicCallback(
        self.clean_old_unacked_orders, self._og_config.unacked_order_check_period * 1000)
    self._unacked_order_periodic_cleaner.start()

  def stop(self):
    self._unacked_order_periodic_cleaner.stop()

    # TODO(jaewon): Make proper clean up.
    raise NotImplementedError()

  def get_working_order(self) -> List[Order]:
    return list(self._working_orders.values())

  def submit_impl(self,
                  product: OkexFuturesProduct,
                  price: float,
                  qty: float,
                  order_side: OkexFuturesOrderSide,
                  order_type: OrderType,
                  leverage: int = 10,
                  order_id: Optional[str] = None,
                  tag=None):
    assert self.is_ready()
    assert order_type == OrderType.LIMIT
    assert order_id is None  # Not supported yet
    assert tag is None

    if self._account_checker_failed:
      # enforce close only
      if order_side not in (OkexFuturesOrderSide.BUY_CLOSE, OkexFuturesOrderSide.SELL_CLOSE):
        return False

    timestamp = get_timestamp()
    order_id = order_id or self._order_id_generator.gen_order_id()
    self._og_logger.gen_order_request_and_order_event_and_log(product.symbol,
                                                              price,
                                                              qty,
                                                              order_side,
                                                              order_type,
                                                              order_id,
                                                              timestamp)

    fill = FillManager(price_orig=price, qty_orig=qty, qty_accum_fill=0, price_avg_fill=0)
    order = Order(order_id=order_id,
                  product=product,
                  price=price,
                  qty=qty,
                  order_side=order_side,
                  order_type=order_type,
                  tag=None,
                  accepted=False,
                  cancel_sent=False,
                  internal=create_okex_futures_order_internal(lever_rate=leverage,
                                                              fill=fill,
                                                              timestamp=timestamp))
    self._working_orders[order_id] = order

    self._ioloop.add_callback(self._submit_impl, order_id, order)

  @gen.coroutine
  def _submit_impl(self, order_id, order):
    exchange_order_id = None
    try:
      response = yield self._rest_client.trade(order.product.order_symbol,
                                               order.product.contract_type,
                                               order.order_side,
                                               order.price,
                                               order.qty)
      if response.status_code != 200:
        self._logger.error('Invalid status code: %s (%s)', response.status_code, response.content)
        if response.status_code in (403, 429):
          self._rest_rate_limiter.set_error(status_code=response.status_code)
      msg = json.loads(response.content)
      exchange_order_id = msg['order_id']
      self._order_counter_success += 1
    except requests.exceptions.RequestException:
      self._logger.exception('RequestException')
    except json.JSONDecodeError:
      self._logger.exception('JsonDecodeException: %s', response.content)
    except (TypeError, KeyError):
      self._logger.error('Wrong order: %s', msg)
    except Exception as e:
      self._logger.error('Unknown exception: %s', e)
      raise e

    timestamp = get_timestamp()
    if exchange_order_id is None:
      if order.accepted:
        self._logger.info('Order already accepted! %s', order.internal.exchange_order_id)
      else:
        self._og_logger.gen_order_response_and_order_event_and_log(order.product.symbol,
                                                                   False,
                                                                   order_id,
                                                                   timestamp)
        try:
          del self._working_orders[order_id]
        except KeyError:
          self._logger.info('Order already removed! %s', exchange_order_id)
    else:
      self._og_logger.gen_order_response_and_order_event_and_log(order.product.symbol,
                                                                 True,
                                                                 order_id,
                                                                 timestamp)
      if exchange_order_id in self._orders_by_exchange_order_id:
        prev_order = self._orders_by_exchange_order_id[exchange_order_id]
        if prev_order.order_id in self._working_orders:
          self._logger.info('Replace %s => %s', prev_order.order_id, order_id)
          self._working_orders.pop(order_id, None)
          self._working_orders.pop(prev_order.order_id, None)
          prev_order.order_id = order_id
          prev_order.internal.timestamp = order.internal.timestamp
          self._working_orders[order_id] = prev_order
        else:
          self._logger.warning('order is dead when rest submit response is received! %s\n%s',
                               order,
                               prev_order)
          del self._working_orders[order_id]
        order = prev_order
      else:
        order.accepted = True
        order.internal.exchange_order_id = exchange_order_id
        self._orders_by_exchange_order_id[exchange_order_id] = order
      order.internal.update_received_timestamp = timestamp
      status = 0
      self._og_logger.gen_order_event_and_log(status, order, timestamp, OrderEvent.ORDER_ACCEPTED)

    self._logger.info("_on_spot_order. submitted_order/rsp_orders: %s/%s",
                      self._order_id_generator.internal_counter,
                      self._order_counter_success)

  def cancel_impl(self, order_id, *, ignore_error=False):
    assert self.is_ready()
    if order_id not in self._working_orders:
      self._logger.error('No order_id=%s' % order_id)
      return

    order = self._working_orders[order_id]
    if not order.accepted:
      self._logger.error('order_id=%s has not been accepted' % order_id)
      return

    timestamp = get_timestamp()
    self._og_logger.gen_cancel_request_and_order_event_and_log(order.product.symbol,
                                                               order.order_id,
                                                               timestamp)

    exchange_order_id = order.internal.exchange_order_id
    if not order.cancel_sent:
      order.cancel_sent = True
      order.internal.cancel_sent_time = timestamp
    ret = self._ws_client.cancel(order.product.order_symbol,
                                 order.product.contract_type,
                                 exchange_order_id)
    self._og_logger.gen_cancel_response_and_log(order.product.symbol, True, order.order_id)
    return ret

  def cancel_all_impl(self, *, ignore_error=False):
    for order_id in [key for key in self._working_orders.keys()]:
      self.cancel(order_id)

  def order_ids_by_product(self, product: OkexFuturesProduct):
    res = []
    for order in self._working_orders.values():
      if order.product == product:
        res.append(order.order_id)
    return res
